package com.yanxu;

import com.yanxu.domain.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author 折戟沉沙铁未销
 * @version V1.0
 * @date 2025-07-13-2025
 * @Description: api 学习
 *  map 算子的转换。
 *  map 算子是一种一对一的转换。
 *      比如，我们可以根据我们的输入，针对每一个输入，做一些特殊的处理，封装成对象等。
 *      将每一个输入做处理，然后转换成我们需要的一个对象
 *
 *
 */
public class Api06_Transform_map {
    public static void main(String[] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //源 -算子
        //从 fromElements 中获取源输入
        DataStreamSource<Event> streamSource = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L),
                new Event("zhangsan", "./cart", 2000L),
                new Event("tom", "./likes", 2000L)
        );
        streamSource.print("from elements >>>");

        //使用map 进行算子转换
        SingleOutputStreamOperator<String> mapStreamOperator = streamSource.map(new MapFunction<Event, String>() {
            @Override
            public String map(Event event) throws Exception {
                String name = event.getName();
                return name;
            }
        });
        mapStreamOperator.print("map >>>");

        /*
        * 输出结果：
            from elements >>>> Event(name=Mary, url=./home, duration=1000)
            map >>>> Mary                                                   -----> 经过转换得到的输出
            from elements >>>> Event(name=Bob, url=./cart, duration=2000)
            map >>>> Bob                                                    -----> 经过转换得到的输出
            from elements >>>> Event(name=zhangsan, url=./cart, duration=2000)
            map >>>> zhangsan                                               -----> 经过转换得到的输出
            from elements >>>> Event(name=tom, url=./likes, duration=2000)
            map >>>> tom                                                    -----> 经过转换得到的输出
        * */

        //对环境进行执行。
        env.execute();
    }
}